package com.qinglei.recoup.longconn.server;


import com.qinglei.recoup.common.domain.RadarMessage;
import com.qinglei.recoup.longconn.entity.Message;
import com.qinglei.recoup.longconn.entity.SendRadarDataEvent;
import com.qinglei.recoup.longconn.handler.DeviceStatusHandler;
import com.qinglei.recoup.longconn.handler.HeartBeatHandler;
import com.qinglei.recoup.longconn.handler.MessageDecoder;
import com.qinglei.recoup.longconn.handler.MessageEncoder;
import com.qinglei.recoup.longconn.handler.UploadDataHandler;
import com.qinglei.recoup.longconn.utils.ProtocolUtils;
import io.netty.channel.Channel;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.timeout.IdleStateHandler;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.net.SocketAddress;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;

@Component
@Slf4j
public class NettyServerInitializer extends ChannelInitializer<SocketChannel> implements DeviceStatusHandler {
    private static Map<SocketAddress, Channel> clientChannelMap = new ConcurrentHashMap<>();
    private static Map<String, Channel> deviceChannelMap = new ConcurrentHashMap<>();

    @Resource
    private UploadDataHandler uploadDataHandler;

    @Resource
    private ApplicationEventPublisher applicationEventPublisher;

    @Value("${recoup.longconn.server.idle-timeout:60}")
    private int idleTimeout;

    @Override
    protected void initChannel(SocketChannel ch) throws Exception {
        ChannelPipeline pipeline = ch.pipeline();
        // 根据协议格式，自定义的解码器
//        pipeline.addLast("headerDecoder", new HeaderDecoder());
//        pipeline.addLast("frameDecoder", new HeaderDecoder(LongConnConstant.MAX_FRAME_LENGTH,
//                LongConnConstant.LENGTH_FIELD_OFFSET, LongConnConstant.LENGTH_FIELD_LENGTH,
//                LongConnConstant.LENGTH_ADJUSTMENT, LongConnConstant.INITIAL_BYTES_TO_STRIP));
//        pipeline.addLast("frameDecoder", new LengthFieldBasedFrameDecoder(LongConnConstant.MAX_FRAME_LENGTH,
//                LongConnConstant.LENGTH_FIELD_OFFSET, LongConnConstant.LENGTH_FIELD_LENGTH,
//                LongConnConstant.LENGTH_ADJUSTMENT, LongConnConstant.INITIAL_BYTES_TO_STRIP));
        // 客户端连接空闲超时
        pipeline.addLast(new IdleStateHandler(0, 0, this.idleTimeout));
        // 数据解码
        pipeline.addLast("decoder", new MessageDecoder());
        // 维护心跳
        pipeline.addLast("heart", new HeartBeatHandler(clientChannelMap, deviceChannelMap, this));
        // 上传数据处理
        pipeline.addLast("uploadDataHandler", uploadDataHandler);
        // 数据编码
        pipeline.addLast("encoder", new MessageEncoder());
    }

//    @Scheduled(fixedDelay = 1000 * 10, initialDelay = 1000 * 5)
//    public void checkChannel(){
//        for(Map.Entry<SocketAddress, Channel> entry : clientChannelMap.entrySet()){
//            SocketAddress key = entry.getKey();
//            Channel channel = entry.getValue();
//            if(!channel.isActive()){
//                clientChannelMap.remove(key);
//                Iterator<Map.Entry<String, Channel>> iterator = deviceChannelMap.entrySet().iterator();
//                while (iterator.hasNext()){
//                    Map.Entry<String, Channel> next = iterator.next();
//                    Channel removeChannel = next.getValue();
//                    if(channel.equals(removeChannel)){
//                        String sn = next.getKey();
//                        deviceChannelMap.remove(sn);
//                        procDeviceStatus(sn, RadarMessage.STATUS_OFFLINE);
//                    }
//                }
//            }
//        }
//    }

    /**
     * 发送下行消息给所有客户端
     * @param funcCode
     * @param body
     * @return
     */
    public int sendMessageAllClient(String funcCode, String body) {
        AtomicInteger deviceCnt = new AtomicInteger();
        deviceChannelMap.forEach((sn, channel) -> {
            Message message = ProtocolUtils.buildDownCtrlMessage(sn, funcCode, body);
            if (channel.isActive()) {
                channel.writeAndFlush(message);
                deviceCnt.getAndIncrement();
                log.info("send to {} message:{}", sn, message);
            }
        });
        return deviceCnt.intValue();
    }
    /**
     * 发送下行消息给指定客户端
     * @param snList
     * @param funcCode
     * @param body
     * @return
     */
    public int sendMessage2Client(List<String> snList, String funcCode, String body) {
        AtomicInteger deviceCnt = new AtomicInteger();
        snList.forEach(sn -> {
            Message message = ProtocolUtils.buildDownCtrlMessage(sn, funcCode, body);
            Channel channel = deviceChannelMap.get(sn);
            if (channel == null) {
                log.warn("sendMessage2Client channel is null, sn:{}", sn);
                return;
            }
            if (channel.isActive()) {
                channel.writeAndFlush(message);
                deviceCnt.getAndIncrement();
                log.info("send to {} message:{}", sn, message);
            } else {
                log.warn("sendMessage2Client channel is inactive, sn:{}, address:{}", sn, channel.remoteAddress());
            }

        });
        return deviceCnt.intValue();
    }

    /**
     * 处理设备状态变化
     * @param sn
     * @param status
     */
    @Override
    public void procDeviceStatus(String sn, Integer status) {
        RadarMessage radarMessage = new RadarMessage();
        radarMessage.setSn(sn);
        radarMessage.setType(RadarMessage.TYPE_DEVICE_STATUS);
        radarMessage.setData(status);
        radarMessage.setTimestamp(System.currentTimeMillis());
        applicationEventPublisher.publishEvent(new SendRadarDataEvent(radarMessage));
    }
}
